import org.apache.spark.sql.{SparkSession, DataFrame}
// 导入隐式转换
import spark.implicits._

object SparkSQLUpdate {
  def main(args: Array[String]): Unit = {
    // 创建SparkSession
    val spark = SparkSession.builder()
      .appName("Spark SQL Update Example")
      .config("spark.master", "local")
      .getOrCreate()

    // 连接MySQL数据库
    val url = "jdbc:mysql://localhost:3306/sparksql_practice"
    val properties = new java.util.Properties()
    properties.setProperty("user", "lsz")
    properties.setProperty("password", "lsz")

    // 读取students表数据
    var studentsDF: DataFrame = spark.read.jdbc(url, "students", properties)

    // 更新一名特定学生的专业为“Data Science”
    val studentIDToUpdate = 1 // 假设要更新ID为1的学生的专业
    studentsDF = studentsDF.withColumn("major", when($"student_id" === studentIDToUpdate, "Data Science").otherwise($"major"))

    // 增加所有学生年龄1岁（模拟年龄增长）
    studentsDF = studentsDF.withColumn("age", $"age" + 1)

    // 将更新后的数据重新写入数据库
    studentsDF.write.mode("overwrite").jdbc(url, "students", properties)

    // 验证更新结果
    val updatedStudentsDF: DataFrame = spark.read.jdbc(url, "students", properties)
    println("更新后的学生信息：")
    updatedStudentsDF.show()

    // 关闭SparkSession
    spark.stop()
  }
}

